package com.xinqing.bigdata.test.datastream

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

/**
  * @Author:CHQ
  * @Date:2020 /6/22 15:32
  * @Description
  */
class MyAssignser() extends AssignerWithPeriodicWatermarks[SensorReading] {

  val bound = 3 * 1000L
  var maxTime = Long.MinValue

  override def getCurrentWatermark: Watermark = {
    new Watermark(maxTime - bound)
  }

  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    maxTime = maxTime.max(t.timestamp * 1000L)
    t.timestamp * 1000L
  }
}
